package me.chan.parallelstream;

import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.RecursiveTask;
import java.util.stream.LongStream;

import org.omg.Messaging.SyncScopeHelper;

@SuppressWarnings("serial")
public class ForkJoinSumCalculator  extends RecursiveTask<Long>{

	private final long[] numbers;                         //子任务处理的数组
	private final int start;										//起始位置
	private final int end;											//终止位置
	
	private static final int THRESHOLD = 10_000;	//不再将数组进行子任务划分的数组大小
	
	public ForkJoinSumCalculator(long[] numbers) {
		this(numbers, 0, numbers.length );
	}
	
	public ForkJoinSumCalculator(long[] numbers, int start, int end) {
		this.numbers = numbers;
		this.start = start;
		this.end = end;
	} 
	
	@Override
	protected Long compute() {
		int length = end - start;
		//如果小于10_000，串行计算
		if (length <= THRESHOLD) {
			return computeSequentially();
		}
		//否则fork出子任务并行j计算
		
		//leftTask 为数组的前半部分值求和
		ForkJoinSumCalculator leftTask = new ForkJoinSumCalculator(numbers, start, start + length/2 );
		leftTask.fork();
		//rightTask 为数组的后半部分值求和
		ForkJoinSumCalculator rightTask = new ForkJoinSumCalculator(numbers, start + length/2 , end);
		Long rightResult = rightTask.compute();
		Long leftResult = leftTask.join();
		return leftResult + rightResult;
	}

	private Long computeSequentially() {
		long sum = 0;
		for (int i=start; i<end; i++) {
			sum += numbers[i];
		}
		return sum;
	}
	
	public static long computeSequence(long n) {
		long[] numbers = LongStream.rangeClosed(0,  n).toArray();
		long sum = 0;
		for (int i=0; i<numbers.length; i++) {
			sum += numbers[i];
		}
		return sum;
	}

	public static long forkJoinSum(long  n) {
		long[] numbers = LongStream.rangeClosed(0, n).toArray();
		ForkJoinTask<Long> task = new ForkJoinSumCalculator(numbers);
		return new ForkJoinPool().invoke(task);
	}
	
	public static void main(String... args) {
		long startTime  = System.nanoTime();
		//System.out.println();
		forkJoinSum(1_000_000);
		System.out.println((System.nanoTime() - startTime) / 1000_000 + "msec");
		
		long sequenceTime = System.nanoTime();
		computeSequence(1_000_000);
		System.out.println((System.nanoTime() - sequenceTime) / 1000_000 + "msec");
	}
}
